package com.shux.kafka.consumer;

import java.util.concurrent.BlockingQueue;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月20日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class PackMessageThread<T> extends RepeatThread{
	private BlockingQueue<KafkaBatchMessage<T>> packMessageQueue;
	private BlockingQueue<KafkaMessage<T>> fetchMessageQueue;
	private ShuxKafkaConsumer<T> kafkaConsumer;
	/**
	 * 当kafka中没有数据时，即队列packMessageQueue中没有数据时，睡眠多少毫秒
	 */
	private int packNoDataSleepMillis;
	/**
	 * 整个群组最多等待的时间,即整个组等待了这个时间，如果还没有接收到messageGroupSize个message，则也退出循环
	 */
	private int waitGroupSizeMillis;
	private int messageGroupSize;
	public PackMessageThread(Integer sleepSecords) {
		super(sleepSecords);
	}
	public PackMessageThread(Integer sleepSecords,
			BlockingQueue<KafkaMessage<T>> fetchMessageQueue,
			BlockingQueue<KafkaBatchMessage<T>> packMessageQueue,
			ConsumerOptionConfig optionConfig,
			ShuxKafkaConsumer<T> kafkaConsumer) {
		super(sleepSecords);
		this.fetchMessageQueue = fetchMessageQueue;
		this.packMessageQueue = packMessageQueue;
		this.kafkaConsumer = kafkaConsumer;
		this.packNoDataSleepMillis = optionConfig.getPackNoDataSleepMillis();
		this.waitGroupSizeMillis = optionConfig.getWaitGroupSizeMillis();
		this.messageGroupSize = optionConfig.getMessageGroupSize();
		
	}
	@Override
	protected void repeatWork() throws InterruptedException {
		logger.debug("准备从fetchMessageQueue中获取数据，并且组成批量数据push到队列packMessageQueue中");
		KafkaBatchMessage<T> batchMessage = new KafkaBatchMessage<>();
		long groupEndTime = System.currentTimeMillis() + waitGroupSizeMillis;
		for(;;){
			KafkaMessage<T> message = fetchMessageQueue.poll();
			//logger.info("当前packMessageQueue大小为：{}",packMessageQueue.size());
			if (message != null) {
				logger.debug("从fetchMessageQueue中取到数据:" + message.getMessage());
				if( batchMessage.getSize() >= messageGroupSize - 1 ){//装满了就退出循环，装满之后会发送给调用者
					batchMessage.addKafkaMessage(message);
					addPackMessageToQueue(batchMessage, packMessageQueue);
					break;
				} else {
					batchMessage.addKafkaMessage(message);
				}
			} else {
				if (System.currentTimeMillis() >= groupEndTime ){
					if(batchMessage.isEmpty()) break;//如果为空则退出，这句不能少，因为不能把null put到BlockingQueue中，否则会报错
					addPackMessageToQueue(batchMessage, packMessageQueue);
					break;
				}
				Thread.sleep(packNoDataSleepMillis);
			}
		}
		
	}
	
	private void addPackMessageToQueue(KafkaBatchMessage<T> batchMessage,BlockingQueue<KafkaBatchMessage<T>> packMessageQueue){
		kafkaConsumer.beforeAddPackMessageToQueue(batchMessage);
		logger.debug("put 消息到packMessageQueue，put前packMessageQueue 大小为{}",packMessageQueue.size());
		QueueUtils.pushDataToQueue(batchMessage, packMessageQueue);
	}

}
